package cn.doitedu.rtmk.demo10;

import cn.doitedu.rtmk.common.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.text.ParseException;

public class Model_X_Calculator implements RuleCalculator {

    // 规则参数
    JSONObject paramJsonObject;

    // 规则人群
    Roaring64Bitmap preSelectUsersBitmap;

    // 跨历史段预统计，截止时间点
    long preTjEndtime;

    // 用于统计动态画像条件事件次数的状态
    ValueState<Integer> cntState;

    String ruleId;

    String realtimeProfileEventId;
    int realtimeProfileEventCnt;
    int flag;
    long endTime;
    String fireEventId;

    JSONObject message;

    Jedis jedis;

    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJsonStr, Roaring64Bitmap preSelectUsersBitmap, long preTjEndtime) throws IOException, ParseException {

        paramJsonObject = JSON.parseObject(ruleParamJsonStr);

        this.preSelectUsersBitmap = preSelectUsersBitmap;

        this.preTjEndtime = preTjEndtime;

        // 申请一个用于统计动态画像条件事件次数的状态
        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("cnt_state", Integer.class));

        // 取出本规则的规则id
        ruleId = paramJsonObject.getString("rule_id");

        // 取出本规则的动态画像条件参数
        JSONArray realtimeProfileJsonArray = paramJsonObject.getJSONArray("realtime_profile");
        JSONObject realtimeProfileParamObject = realtimeProfileJsonArray.getJSONObject(0);

        // 取出动态画像条件中的各种参数
        realtimeProfileEventId = realtimeProfileParamObject.getString("eventId");  // 目标事件
        realtimeProfileEventCnt = realtimeProfileParamObject.getInteger("eventCnt");  // 目标次数
        flag = realtimeProfileParamObject.getInteger("flag");  // 条件标识

        String realtimeProfileEndTime = realtimeProfileParamObject.getJSONArray("timeRange").getString(1);
        endTime = DateUtils.parseDate(realtimeProfileEndTime, "yyyy-MM-dd HH:mm:ss").getTime();  // 目标统计结束时间

        // 触发条件事件
        fireEventId = paramJsonObject.getString("fire_action");

        // 创建redis客户端连接
        jedis = new Jedis("doitedu", 6379);

        // 用于输出信息的对象
        message = new JSONObject();
        message.put("rule_id", ruleId);


    }

    @Override
    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {

        // 如果该行为用户不是预圈选的目标人群，则啥也不做
        if (!preSelectUsersBitmap.contains(eventBean.getUid())) return;


        // 如果收到的数据是动态画像统计的目标事件
        if (eventBean.getEventId().equals(realtimeProfileEventId)
                && eventBean.getTimestamp() > preTjEndtime - 8*60*60*1000
                && eventBean.getTimestamp() < endTime) {

            // 看内部状态中，该用户该目标事件，是否已经有值；，否则直接在内部值上＋1
            Integer stateCnt = cntState.value();

            // 如果没有值，需要去redis取初始值
            int redisCnt = 0;
            if (stateCnt == null) {
                // redis中的数据结构： hash
                // 大key：  规则id:条件id
                // 小key：  userId
                String redisCntStr = jedis.hget(ruleId + ":" + flag, eventBean.getUid() + "");
                if (redisCntStr != null) {
                    redisCnt = Integer.parseInt(redisCntStr);
                }
            }

            // 将redis获取到的值，放入内部状态
            cntState.update(redisCnt + 1);
        }

        // 如果收到的是触发条件目标事件
        if (eventBean.getEventId().equals(fireEventId)) {

            Integer stateValue = cntState.value();

            int redisCnt = 0;
            if (stateValue == null) {
                // redis中的数据结构： hash
                // 大key：  规则id:条件id
                // 小key：  userId
                String redisCntStr = jedis.hget(ruleId + ":" + flag, eventBean.getUid() + "");
                if (redisCntStr != null) {
                    redisCnt = Integer.parseInt(redisCntStr);
                }

                cntState.update(redisCnt);
            }

            stateValue = cntState.value();


            if (stateValue >= realtimeProfileEventCnt) {

                message.put("user_id", eventBean.getUid());
                message.put("match_time", eventBean.getTimestamp());

                collector.collect(message.toJSONString());

            }

        }
    }
}
